摘要
這篇文章探討了 LangChain 和 LangGraph 這兩個強大的工具,它們能夠賦予 AI 模型呼叫外部程式碼的能力,進而擴展其功能並實現更智能的交互。文章首先介紹了「工具」的概念,並說明其如何充當 AI 模型與外部世界溝通的橋樑。接著,文章深入解析了在 LangChain 中創建工具的各種方法,從簡單的函數裝飾器到更靈活的 StructuredTool,並強調錯誤處理在確保 AI 應用穩健性的重要性。文章最後引入 LangGraph,展示如何使用 ToolNode 來整合外部工具,並通過一個簡單的聊天機器人範例說明了其實際應用。透過本文的探討,我們可以學習到如何利用 LangChain 和 LangGraph 打造功能更強大、互動更自然的 AI 應用。
在人工智慧和自然語言處理領域中,LangChain 和 LangGraph 是兩個備受矚目的工具。本文將深入探討如何讓 AI 模型呼叫您的程式,擴展其功能並實現更智能的交互。
Tools(工具)是專門設計給 AI 模型使用的功能單元。它們接受模型生成的輸入,並將輸出回傳給模型使用。當您希望模型能夠控制您程式碼的某些部分,或是呼叫外部 API 時,Tools 就顯得尤為重要。
一個典型的 Tool 通常包含以下幾個關鍵元素:
當 Tool 綁定到模型時,其名稱、描述和 JSON 結構會作為上下文提供給模型。給定一系列工具和一組指令,模型可以請求使用特定輸入呼叫一個或多個工具。
讓我們來看一個簡單的例子,假設我們要創建一個查詢台灣天氣的工具:
@tool
def query_taiwan_weather(city: str, date: Optional[str] = None) -> str:
"""
查詢台灣特定城市的天氣情況。
參數:
city (str): 要查詢天氣的城市名稱,例如 "台北"、"高雄" 等。
date (str, 可選): 要查詢的日期,格式為 "YYYY-MM-DD"。如果不提供,則查詢當天天氣。
返回:
str: 包含天氣信息的字符串。
"""
# 這裡應該是實際的天氣 API 調用邏輯
# 為了示例,我們返回一個模擬的結果
return f"{city} 在 {date or '今天'} 的天氣晴朗,溫度 25°C,適合外出活動。"
讓我們看看這個工具的 JSON Schema:
rich.print(query_taiwan_weather.args_schema.schema())
現在,我們來使用這個工具,並查看返回的訊息結構:
from langchain_openai import ChatOpenAI
llm = ChatOpenAI(model="gpt-3.5-turbo-0125")
llm_with_weather = llm.bind_tools([query_taiwan_weather])
response = llm_with_weather.invoke("請告訴我台北明天的天氣如何?")
rich.print(response)
🌟 亮點:通過使用 Tools,我們可以大大擴展 AI 模型的能力,使其能夠執行更加複雜和具體的任務,同時保持靈活性和可控性。
現在我們已經了解了 Tools 在 LangChain 中的角色,讓我們深入探討如何創建這些 Tools。當您在構建一個 agent(代理)時,提供一系列它可以使用的 Tools 是至關重要的。
以下是構成 Tool 的主要屬性:
屬性 | 型別 | 說明 |
---|---|---|
name | 字串 | 必須在提供給 LLM 或代理的一組工具中是唯一的。 |
description | 字串 | 描述這個工具的功能。作為 LLM 或代理的上下文使用。 |
args_schema | Pydantic BaseModel | 選擇性但建議使用,可用於提供更多資訊或對預期參數進行驗證。 |
return_direct | 布林值 | 僅與代理相關。當設為 True 時,在調用給定工具後,代理會停止並直接將結果返回給使用者。 |
LangChain 支持以下幾種創建 Tools 的方法:
💡 專業提示:模型的表現會因 Tools 的名稱、描述和 JSON 結構的選擇而有所不同。精心設計這些元素可以顯著提升模型的效能。
@tool
裝飾器提供了一種極為便捷的方式來定義和創建自訂工具。這個裝飾器的設計理念是讓開發者能夠快速將普通的 Python 函式轉變為 LangChain 可識別和使用的工具。
主要特點:
📌 注意事項: 確保為每個使用 @tool 裝飾器的函式提供清晰、詳細的文件字串,以便 LLM 或代理能夠正確使用該工具。
以查詢台灣城市天氣為例,展示 @tool 的基本用法:
from langchain.tools import tool
@tool
def get_taiwan_weather(city: str) -> str:
"""查詢台灣特定城市的天氣狀況。"""
weather_data = {
"台北": "晴天,溫度28°C",
"台中": "多雲,溫度26°C",
"高雄": "陰天,溫度30°C"
}
return f"{city}的天氣:{weather_data.get(city, '暫無資料')}"
# 檢視工具的相關屬性
print(f"name: {get_taiwan_weather.name}")
print(f"description: {get_taiwan_weather.description}")
print(f"args: {get_taiwan_weather.args}")
# 使用這個工具
result = get_taiwan_weather.run("台北")
print(f"呼叫工具結果{result}")
在這個例子中:
對於可能需要等待外部 API 回應的情況,我們可以使用非同步版本的工具:
import asyncio
@tool
async def get_taiwan_weather_async(city: str) -> str:
"""非同步查詢台灣特定城市的天氣狀況。"""
await asyncio.sleep(1) # 模擬 API 調用
weather_data = {
"台北": "晴天,溫度28°C",
"台中": "多雲,溫度26°C",
"高雄": "陰天,溫度30°C"
}
return f"{city}的天氣:{weather_data.get(city, '暫無資料')}"
result = await get_taiwan_weather_async.ainvoke({"city": "高雄"})
print(result)
這個非同步版本的工具:
invoke
來呼叫 async
異步工具🚀 效能提示:非同步工具在處理需要等待外部資源(如網路請求)的場景中特別有用,可以提高整體應用的效能。
更多內容可以參考 Langchain/How-to-create-async-tools
有時,我們可能想要更精確地控制工具的名稱和參數結構。這可以通過向 @tool 裝飾器傳遞參數來實現:
from langchain.pydantic_v1 import BaseModel, Field
class TravelPlanInput(BaseModel):
city: str = Field(description="欲訪問的台灣城市")
days: int = Field(description="旅遊天數")
budget: int = Field(description="預算(新台幣)")
@tool("台灣旅遊規劃器", args_schema=TravelPlanInput, return_direct=True)
def plan_taiwan_trip(city: str, days: int, budget: int) -> str:
"""根據指定的城市、天數和預算規劃台灣旅遊行程。"""
per_day_budget = budget // days
if city == "台北":
return f"台北{days}天行程:每天預算{per_day_budget}元。建議景點:101大樓、故宮博物院、象山。別忘了品嚐鼎泰豐的小籠包!"
elif city == "台中":
return f"台中{days}天行程:每天預算{per_day_budget}元。建議景點:高美濕地、彩虹眷村、逢甲夜市。記得嘗試太陽餅!"
elif city == "高雄":
return f"高雄{days}天行程:每天預算{per_day_budget}元。建議景點:蓮池潭、駁二藝術特區、旗津海岸。不要錯過壽山動物園!"
else:
return f"抱歉,目前沒有{city}的旅遊規劃資訊。"
print(plan_taiwan_trip.name)
print(plan_taiwan_trip.description)
print(plan_taiwan_trip.args)
print(plan_taiwan_trip.return_direct)
# 使用這個工具
result = plan_taiwan_trip.run({"city": "台北", "days": 3, "budget": 10000})
print(result)
在這個例子中:
這種方法允許我們對工具的結構和行為有更精細的控制,特別適用於複雜的工具或需要特定輸入驗證的場景。
StructuredTool.from_function
類方法提供了比 @tool
裝飾器更多的配置選項,同時不需要太多額外的程式碼。這個部分將展示如何使用 StructuredTool.from_function
方法來配置和使用工具,同時融入台灣的夜市文化元素。
首先,讓我們創建一個簡單的夜市小吃價格計算器:
from langchain_core.tools import StructuredTool
def calculate_night_market_price(item: str, quantity: int) -> str:
"""計算台灣夜市小吃的總價。"""
prices = {
"蚵仔煎": 60,
"臭豆腐": 40,
"珍珠奶茶": 50,
"鹽酥雞": 70,
"大腸包小腸": 55
}
if item not in prices:
return f"抱歉,我們沒有 {item} 的價格信息。"
total = prices[item] * quantity
return f"{quantity} 份 {item} 的總價是 {total} 元新台幣。"
async def acalculate_night_market_price(item: str, quantity: int) -> str:
"""非同步計算台灣夜市小吃的總價。"""
return calculate_night_market_price(item, quantity)
night_market_calculator = StructuredTool.from_function(
func=calculate_night_market_price,
coroutine=acalculate_night_market_price
)
print(night_market_calculator.invoke({"item": "臭豆腐", "quantity": 2}))
print(await night_market_calculator.ainvoke({"item": "珍珠奶茶", "quantity": 3}))
StructuredTool 的一個主要優勢是它能夠同時支援同步(sync)和非同步(async)函數。
這樣配置允許工具根據調用方式(invoke 或 ainvoke)自動選擇適當的函數。
💡 專業提示:使用 StructuredTool 可以讓您的工具更加靈活,能夠應對不同的執行環境和需求。
更多內容可以參考官方文件
現在,讓我們看看如何進一步配置 StructuredTool
:
from langchain.pydantic_v1 import BaseModel, Field
class NightMarketInput(BaseModel):
item: str = Field(description="夜市小吃名稱,例如:蚵仔煎、臭豆腐、珍珠奶茶等")
quantity: int = Field(description="購買數量")
def calculate_night_market_price(item: str, quantity: int) -> str:
"""計算台灣夜市小吃的總價。"""
prices = {
"蚵仔煎": 60,
"臭豆腐": 40,
"珍珠奶茶": 50,
"鹽酥雞": 70,
"大腸包小腸": 55
}
if item not in prices:
return f"抱歉,我們沒有 {item} 的價格信息。"
total = prices[item] * quantity
return f"{quantity} 份 {item} 的總價是 {total} 元新台幣。"
night_market_calculator = StructuredTool.from_function(
func=calculate_night_market_price,
name="台灣夜市小吃計價器",
description="計算台灣夜市常見小吃的總價",
args_schema=NightMarketInput,
return_direct=True
)
print(night_market_calculator.invoke({"item": "蚵仔煎", "quantity": 2}))
print(night_market_calculator.name)
print(night_market_calculator.description)
print(night_market_calculator.args)
進階配置中,可以清楚看到以下特性:
定義輸入模型:
自定義工具名稱和描述:
指定參數結構:
直接返回結果:
🌟 亮點:StructuredTool 為開發者提供了更大的靈活性,特別適合需要精確控制輸入輸出結構的複雜應用場景。
在使用 LangChain 的工具(Tools)時,建立一個錯誤處理策略是非常重要的,特別是當這些工具與代理(Agents)一起使用時。適當的錯誤處理可以讓代理從錯誤中恢復並繼續執行。讓我們深入探討如何在 LangChain 中實現工具的錯誤處理。
一個簡單的錯誤處理策略是在工具內部拋出 ToolException
,並使用 handle_tool_error
來指定錯誤處理方式。當指定了錯誤處理器時,異常會被捕獲,錯誤處理器將決定從工具返回什麼輸出。
您可以將 handle_tool_error
設置為以下幾種方式:
⚠️ 重要提示:僅僅拋出
ToolException
是不夠的。您需要先設置工具的 handle_tool_error,因為其預設值為 False。
讓我們通過一個查詢天氣的例子來說明這些錯誤處理方法:
def get_weather(city: str) -> int:
"""查詢指定城市的天氣。"""
raise ToolException(f"錯誤:找不到名為 {city} 的城市。")
from langchain_core.tools import ToolException
from langchain.tools import StructuredTool
# 範例 1:預設錯誤處理行為
get_weather_tool = StructuredTool.from_function(
func=get_weather,
handle_tool_error=True,
)
result = get_weather_tool.invoke({"city": "未知城市"})
print("預設錯誤處理:", result)
預設錯誤處理:
# 範例 2:使用自定義字串
get_weather_tool = StructuredTool.from_function(
func=get_weather,
handle_tool_error="抱歉,找不到該城市,但那裡的溫度應該高於絕對零度!",
)
result = get_weather_tool.invoke({"city": "未知城市"})
print("自定義字串錯誤處理:", result)
自定義字串錯誤處理:
handle_tool_error
為一個字串,會在發生錯誤時始終返回該字串。# 範例 3:使用自定義函數
def custom_error_handler(error: ToolException) -> str:
return f"工具執行期間發生以下錯誤:`{error.args[0]}`"
get_weather_tool = StructuredTool.from_function(
func=get_weather,
handle_tool_error=custom_error_handler,
)
result = get_weather_tool.invoke({"city": "未知城市"})
print("自定義函數錯誤處理:", result)
自定義函數錯誤處理:
🛠️ 最佳實踐:根據您的應用需求選擇適當的錯誤處理方法。對於簡單情況,使用預設或自定義字串可能就足夠了。對於需要更複雜邏輯的情況,自定義函數提供了最大的靈活性。
在前面的章節中,我們深入探討了 LangChain 中工具(Tool)的各個方面,包括創建方法和異常處理等。現在,讓我們將焦點轉向 LangGraph,看看如何在這個強大的框架中運用這些知識。本文將通過一個簡單而實用的例子,逐步展示 LangGraph 的操作方法。
為了增強聊天機器人的能力,我們將整合一個網絡搜索工具。這使得機器人能夠查找相關信息,提供更準確的回答。
from langchain_community.tools.tavily_search import TavilySearchResults
from langgraph.prebuilt import ToolNode, tools_condition
tool = TavilySearchResults(max_results=2)
tools = [tool]
tool_node = ToolNode(tools=[tool])
tool.invoke("高雄必比登推薦餐廳?")
ToolNode 封裝了我們定義的工具(在這個例子中是網路搜尋工具)。它允許聊天機器人在需要時調用外部資源來增強其回應能力
將所有組件組合成一個完整的圖形結構。這定義了聊天機器人的工作流程。
直接把剛剛建立的 tool_node
放盡 Graph 當中。
graph_builder = StateGraph(State)
graph_builder.add_node("chatbot", chatbot)
graph_builder.add_node("tools", tool_node)
同時也添加條件邊,讓 chatbot 節點輸出結果中帶有 tool_calls
前往指定節點
graph_builder.add_conditional_edges(
"chatbot",
tools_condition,
)
最後,我們創建一個簡單的交互界面,允許用戶與聊天機器人對話。
while True:
user_input = input("User: ")
if user_input.lower() in ["quit", "exit", "q"]:
print("Goodbye!")
break
for event in graph.stream({"messages": ("user", user_input)}):
for value in event.values():
# print("Assistant:", value["messages"][-1].content)
value["messages"][-1].pretty_print()
這個循環允許用戶輸入消息,聊天機器人會處理這些輸入並給出回應。用戶可以隨時輸入 "quit"、"exit" 或 "q" 來結束對話。
通過本文的探討,我們深入了解了 LangChain 和 LangGraph 中 Tools 的重要性和實現方法。從基本的 @tool 裝飾器到更為靈活的 StructuredTool,再到錯誤處理策略,這些知識將幫助您構建更加強大和可靠的 AI 應用。
記住,選擇合適的工具創建方法和錯誤處理策略,不僅可以提高您的 AI 模型的功能性,還能大大提升其穩定性和用戶體驗。在實際應用中,根據具體需求靈活運用這些技術,將使您的 AI 解決方案更加出色。
最後,隨著 AI 技術的不斷發展,保持學習和實踐的態度至關重要。希望本文能為您的 AI 開發之旅提供有價值的見解和指導。
本篇教學程式碼位於比賽用 Repo,記得多多操作
6. 參考資料:
1.Lang Chain Tool 官方文件
2.Lang Chain Tool 觀念
3.LangChain Tool_calling
4.LangChain Tool_calling觀念
5.LangChain Tool_calling長篇Blog
6.LangChain Tool_calling結合Message
7.LangGRaph Tools Node:
9. Handle Tools node errors:
請問文章中第四節 "LangGraph 如何善用 ToolNode" 中的範例算是 ReAct Agent 嗎?
這跟 LangGraph 的 create_react_agent 有什麼不同?